package com.lizhiyu.flink.demo5_window;

import com.lizhiyu.flink.demo2_source.CustomSource2;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 *  滑动窗口  和 滚动窗口
 *  滑动窗口   SlidingProcessingTimeWindows     例如 5秒内查询 近15秒内的数据汇总
 *  滚动窗口   TumblingProcessingTimeWindows    例如 5秒内查询5秒内的数据
 *            countWindow                     指定时间超过多少个则触发
 *  数据收集是左闭右开的，当达到窗口时间才会将数据数据
 */
public class WindowDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> ds = env.addSource(new CustomSource3());
        ds.setParallelism(1);
        ds.print("------------source:");
        SingleOutputStreamOperator<Tuple3<String, Integer,String>> mapDs = ds.map(new MapFunction<String, Tuple3<String, Integer,String>>() {
            @Override
            public Tuple3<String, Integer,String> map(String value) throws Exception {
                String[] split = value.split(":");
                Tuple3 Tuple3 = new Tuple3(split[0],Integer.parseInt(split[1]),split[2]);
                return Tuple3;
            }
        });

        KeyedStream<Tuple3<String, Integer,String>, String> keyByDs = mapDs.keyBy(new KeySelector<Tuple3<String, Integer,String>, String>() {
            @Override
            public String getKey(Tuple3<String, Integer,String> value) throws Exception {
                return value.f0;
            }
        });
        //sum是对tuple中指定的字段进行求和
        //在指定时间进行一次汇总  3三秒内打印一下三秒内的数据
//        SingleOutputStreamOperator<Tuple3<String, Integer>> sumDS = keyByDs.window(TumblingProcessingTimeWindows.of(Time.seconds(3))).sum(1);
        //每隔三秒打印一下15秒内的数据
//        SingleOutputStreamOperator<Tuple3<String, Integer>> sumDS = keyByDs.window(SlidingProcessingTimeWindows.of(Time.seconds(15),Time.seconds(3))).sum(1);


        //分组后的组内数据超过5个则触发
        //DataStream<VideoOrder> sumDS = keyByDS.countWindow(5).sum("money");
        //分组后的组内数据超过3个则触发统计过去的5个数据
        SingleOutputStreamOperator<Tuple3<String, Integer,String>> sumDS = keyByDs.countWindow(5, 3).sum(1);

        sumDS.print("sink:");
        env.execute();
    }
}
